package com.atguigu.financial.lease.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.financial.lease.realtime.common.FinancialLeaseCommon;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import javax.print.attribute.standard.MediaSize;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;


public class HBaseUtil {

    public static void main(String[] args) {
//        Connection connection = getHBaseConnection();
        try {
            AsyncConnection asyncConnection = getHBaseAsyncConnection();
//            createTable(connection,"default", "t1","0");
//            deleteTable(connection,"default", "t2");
//            closeHBaseConnection(connection);
            JSONObject object = asyncReadDim(asyncConnection, FinancialLeaseCommon.HBASE_NAMESPACE, "dim_industry", "16");
            System.out.println(object.toJSONString());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void loginKerberos(Configuration conf ){
        String filePath ="C:\\ProgramData\\MIT\\Kerberos5";
        System.setProperty("java.security.krb5.conf", filePath + "\\krb5.ini");

        conf.set("hadoop.security.authentication", "kerberos");

        UserGroupInformation.setConfiguration(conf);
        try {
            // hadoop kerberos 认证
            UserGroupInformation.loginUserFromKeytab("hbase-bigdata@TSSJ.COM", "C:\\ProgramData\\MIT\\Kerberos5\\hbase.headless.keytab");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建hbase连接
     * @return
     */
    public static Connection getHBaseConnection(){

        Configuration conf = HBaseConfiguration.create();
        loginKerberos(conf);


        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM,FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM_HOST);
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT,FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT_VALUE);

        conf.set("hbase.master.kerberos.principal", "hbase/_HOST@TSSJ.COM");
        conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@TSSJ.COM");
        conf.set("hbase.security.authentication", "kerberos");
        conf.set("zookeeper.znode.parent", "/hbase-secure");

        try {
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;

    }

    /**
     * 关闭hbase的连接
     * @param connection
     */
    public static void closeHBaseConnection(Connection connection){
        if (connection!=null && !connection.isClosed()){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void createTable(Connection hBaseConnection, String hbaseNamespace, String tableName, String... familyNames) throws IOException {
        // 判断列族是否存在
        if (familyNames == null || familyNames.length==0){
            throw new RuntimeException("列族至少有一个");
        }
        // hbase的api分两类DDL和DML
        // DDL 使用的是admin
        // DML 使用的是table
        Admin admin = hBaseConnection.getAdmin();
        TableName tableName1 = TableName.valueOf(hbaseNamespace, tableName);
        try {
            // 判断表格是否已经存在
            if (admin.tableExists(tableName1)){
                System.out.println("想要创建的表格" + hbaseNamespace + ":" + tableName + "已经存在了");
            }else{
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName1);
                for (String familyName : familyNames) {
                    ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName));
                    tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
                }

                admin.createTable(tableDescriptorBuilder.build());
            }
        }catch (Exception e){
            e.printStackTrace();
        }

        admin.close();

    }

    public static void deleteTable(Connection hBaseConnection, String hbaseNamespace, String tableName) throws IOException {
        Admin admin = hBaseConnection.getAdmin();
        TableName tableName1 = TableName.valueOf(hbaseNamespace, tableName);
        try {
            // 删除表格之前 需要先disable表格
            admin.disableTable(tableName1);
            admin.deleteTable(tableName1);
        }catch (IOException e){
            e.printStackTrace();
        }
        admin.close();
    }

    public static void deleteRow(Connection hBaseConnection, String hbaseNamespace, String sinkTable, String rowKeyValue) throws IOException {
        Table table = hBaseConnection.getTable(TableName.valueOf(hbaseNamespace, sinkTable));

        try {
            table.delete(new Delete(Bytes.toBytes(rowKeyValue)));
        } catch (IOException e) {
            e.printStackTrace();
        }

        table.close();

    }

    public static void putRow(Connection hBaseConnection, String hbaseNamespace, String sinkTable, String rowKeyValue, String sinkFamily, String[] columns, String[] values) throws IOException {
        Table table = hBaseConnection.getTable(TableName.valueOf(hbaseNamespace, sinkTable));
        Put put = new Put(Bytes.toBytes(rowKeyValue));
        for (int i = 0; i < columns.length; i++) {
            if (values[i] == null){
                values[i] = "";
            }
            put.addColumn(Bytes.toBytes(sinkFamily),Bytes.toBytes(columns[i]),Bytes.toBytes(values[i]));
        }
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }

        table.close();
    }

    public static AsyncConnection getHBaseAsyncConnection() throws ExecutionException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        loginKerberos(conf);


        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM,FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM_HOST);
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT,FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT_VALUE);

        conf.set("hbase.master.kerberos.principal", "hbase/_HOST@TSSJ.COM");
        conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@TSSJ.COM");
        conf.set("hbase.security.authentication", "kerberos");
        conf.set("zookeeper.znode.parent", "/hbase-secure");

        return ConnectionFactory.createAsyncConnection(conf).get();
    }

    public static void closeHBaseAsyncConnection(AsyncConnection conn){
        if (conn != null && !conn.isClosed()){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 异步读取hbase中的维度数据
     * @param hBaseAsyncConnection
     * @param hbaseNamespace
     * @param tableName
     * @param rowKey
     */
    public static JSONObject asyncReadDim(AsyncConnection hBaseAsyncConnection, String hbaseNamespace, String tableName, String rowKey) {
        AsyncTable<AdvancedScanResultConsumer> table = hBaseAsyncConnection.getTable(TableName.valueOf(hbaseNamespace, tableName));

        Get get = new Get(Bytes.toBytes(rowKey));
        CompletableFuture<Result> resultFuture = table.get(get);
        JSONObject dim = new JSONObject();
        try {
            Result result = resultFuture.get();

            for (Cell cell : result.rawCells()) {
                String column = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell));
                dim.put(column,value);
            }
            return dim;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("读取维度数据异常");
        }
        return dim;
    }

}
